// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jingyuan

#include "experimental/jingyuan/basis_exchange_strategy.h"

#include <algorithm>
#include <cmath>
#include <exception>
#include <fstream>
#include <memory>
#include <stdexcept>
#include <string>
#include <unordered_map>

#include <boost/algorithm/string/case_conv.hpp>
#include <fmt/core.h>
#include <glog/logging.h>
#include <google/protobuf/stubs/logging.h>
#include <nlohmann/json.hpp>

#include "coin2/base/log.h"
#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/strategy/strategy.h"
#include "ctime"
#include "experimental/bits/fs_path.h"

namespace coin2::strategy::basis_strategy {

constexpr double BPS = 1e-4;
constexpr int64_t NS = 1'000'000'000LL;
constexpr int64_t LOGGING_INTERVAL = 30'000'000'000LL;
constexpr double EPS = 1e-7;  // epsilon for position

// TODO(jingyuan) : warp this inside a class to avoid string construction for same product like
// ProductEncyclopedia

std::string GetProductKey(const IProduct* product) {
  return fmt::format("{}.{}.{}",product->market_name(), product->exchange_name(), product->relative_norm());
}

BasisSubStrategy::BasisSubStrategy(
    BasisStrategy* master_strategy,
    const nlohmann::json& prod_dict,
    const PassiveExecutorConfig& exe_config,
    std::unique_ptr<IProduct> product,
    int64_t timestamp)
    : position_(0),
      max_lean_bps_(prod_dict["max_lean_bps_per_exchange"].get<double>() * BPS),
      price_push_bps_(prod_dict["price_push_bps"].get<double>() * BPS),
      price_pull_bps_(prod_dict["price_pull_bps"].get<double>() * BPS),
      is_spot_(prod_dict["is_spot"].get<bool>()),
      master_strategy_(master_strategy),
      exchange_(MarketExchangeApi::FromString(exe_config.mea()).exchange),
      strategy_key_(GetProductKey(product.get())),
      driver_(nullptr),
      exe_config_(exe_config),
      product_(std::move(product)),
      timestamp_(timestamp) {
  if (prod_dict.contains("starting_position")) {
    simulation_starting_position_ = prod_dict["starting_position"].get<double>();
    position_ = simulation_starting_position_;
  } else {
    simulation_starting_position_ = 0;
  }

  max_position_ = prod_dict["max_position"].get<double>();
  if (prod_dict.contains("min_position")) {
    // explicit stated min_position
    min_position_ = prod_dict["min_position"].get<double>();
  }else{
    //no min position, assumed it's futures and -max_position is the min
    min_position_ = - prod_dict["max_position"].get<double>();
  }

  exe_config_.set_min_pos(min_position_);
  exe_config_.set_max_pos(max_position_);

  previous_max_ = max_position_;
  previous_min_ = min_position_;

  if (is_spot_) {
    lot_size_ = exe_config_.lot_size();
    contract_value_ = prod_dict["contract_value"].get<double>();
    CHECK_GT( contract_value_, 0.0 ) << "contract_value invalid";
    // set lot size to 0 to prevent funny values, it will be updated later
    exe_config_.set_lot_size(0);
  }
}

void BasisSubStrategy::Init(Driver* driver) {
  driver_ = driver;
  auto get_product_holder = [this](const coin2::exchange::base::symbology::IProduct& product, int64_t timestamp) {
      return driver_->product_holder_manager()->GetProductHolder(product, timestamp);
  };

  order_executor_ = std::make_unique<PassiveOrderExecutor>(exe_config_, order_system(), timestamp_, nullptr, get_product_holder);
  // make a copy of the config
  // append the name of the exchange to avoid logger conflict
  reporter_config_ = driver->strategy()->config();
  std::string* mutable_name = reporter_config_.mutable_strategy_name();
  bool name_initialized = false;
  if (std::string::size_type pos = mutable_name->find("{Name}"); pos != std::string::npos) {
    mutable_name->replace(pos, 6, ExchangeType_Name(exchange_));
    name_initialized = true;
  }

  if (std::string::size_type pos = mutable_name->find("{name}"); pos != std::string::npos) {
    std::string exchange_name = ExchangeType_Name(exchange_);
    boost::to_lower(exchange_name);
    mutable_name->replace(pos, 6, exchange_name);
    name_initialized = true;
  }
  CHECK_THROW(name_initialized);
  LOG(INFO) << "Sub Strategy init, config:\n" << reporter_config_.DebugString();
  std::unordered_map<std::string, double>& reserve = master_strategy_->GetReserve(reporter_config_);
  reserve[product_->base().currency()] = (max_position_ + min_position_) / 2.0;

  LOG(INFO) << order_executor_->product().absolute_norm();
}

void BasisSubStrategy::onBookFeed(const FeedUpdate& upd) {
  bool position_updated = false;
  try {
    if (position_ != simulation_starting_position_ + order_executor_->GetPosition()) {
      master_strategy_->SyncPositionAndRisk(upd.timestamp());
      LOG(ERROR) << "This should not happen, missing fill update?";
    }
  } catch (const std::out_of_range& e) {  // Guard against error in sim
    LOG(INFO) << "order executor position not ready" << e.what();
    return;
  }
  if (const auto& result = master_strategy_->fair_value(strategy_key_); result) {
    auto& book = upd.book();
    auto [fair_bid, fair_ask] = result.value();
    double bid = book.Bid0()->price;
    double ask = book.Ask0()->price;
    const auto & product_info = master_strategy_->product_cache.holder(*product_);
    const auto & tick_price = product_info.product_info().tick_price();

    //udpate lot_size
    if (is_spot_){
      double mid = (bid + ask) / 2;
      double exposure = contract_value_ * lot_size_;
      double new_lot_size = exposure / mid;
      LOG_EVERY_N(INFO, 100) << fmt::format(
        "mid {} exp {} new_lot_size {}",
        mid, exposure, new_lot_size);
      order_executor_->UpdateLotSize(new_lot_size);
    }

    double lean_bp = (1 - 2.0 * (position_ - min_position_) / (max_position_ - min_position_) ) * max_lean_bps_;
    double lean = 0.5 * (fair_bid + fair_ask) * lean_bp;

    fair_bid += lean;
    fair_ask += lean;
    // TODO(Jingyuan) round to tick

    if (bid > fair_ask) {
      LOG_EVERY_N(INFO, 100) << fmt::format(
          "Limit the ask {} to {}",
          fair_ask,
          tick_price.GetNextPrice(bid));
      fair_ask = tick_price.GetNextPrice(bid);
    }

    if (ask < fair_bid) {
      LOG_EVERY_N(INFO, 100) << fmt::format(
          "Limit the bid {} to {}",
          fair_bid,
          tick_price.GetPrevPrice(ask));
      fair_bid = tick_price.GetPrevPrice(ask);
    }

    LOG_EVERY_N(INFO, 1000) << fmt::format(
        "lean {} position: {} fair_ask {} fair_bid: {}",
        lean,
        position_,
        fair_ask,
        fair_bid);

    order_executor_->ManageMmOrders(
        upd.timestamp(),
        fair_ask,
        fair_bid,
        0,
        0,
        /*
        fair_ask * (1 - price_pull_bps_),  // sell_price_pull,
        fair_bid * (1 + price_pull_bps_),  // buy_price_pull,
        fair_ask * (1 + price_push_bps_),  // sell_price_push,
        fair_bid * (1 - price_push_bps_),  // buy_price_push,
        */
        false,                             // bool cancel_all_existing_sell,
        false);                            // bool cancel_all_existing_buy
    if (position_updated) {
      ManageTotalRisk(upd.timestamp());
    }

    if (fire_aggressive_order_) {
      // TODO(jingyuan): fix fire_qty for spot
      CHECK(!is_spot_) << "fire_qty not handled properly for spot";

      // TODO(jingyuan): This logic is subject to delay in feed since it submit
      // aggressive order on next book update
      double fire_qty = master_strategy_->GetPosition();
      double fire_price = (bid + ask) * 0.5;
      double total_qty = 0;
      LOG(INFO) << fmt::format(
          "{} will fire aggressive order for {}",
          strategy_key_,
          fire_qty);
      if (abs(fire_qty) < EPS) {
        LOG(INFO) << fmt::format("Position is 0, ignored");
      } else if (fire_qty > 0) {
        // Aggressive sell
        for (auto [price, qty] : book.Bids()) {
          LOG(INFO) << fmt::format("Bid Book price={} qty={}", price, qty);
          if (price < fair_bid) {
            LOG(ERROR) << fmt::format(
                "potential self trade detected bidBook{}/{}, fair_bid={} "
                "request qty={}, current qty{}",
                price,
                qty,
                fair_bid,
                fire_qty,
                total_qty);
            break;
          }
          total_qty += qty;
          fire_price = price;
          if (total_qty >= fire_qty) break;
        }

        if (total_qty > 0) {
          LOG(INFO) << fmt::format("Aggressive SELL order {}/{}", fire_price, fire_qty);
          order_executor_->ForceFireSellOrder(upd.timestamp(), fire_price, fire_qty, "");
        } else {
          LOG(ERROR) << "our order is already top of book, not firing";
        }
      } else {
        // Aggressive buy
        fire_qty = abs(fire_qty);
        for (auto [price, qty] : book.Asks()) {
          LOG(INFO) << fmt::format("Ask Book price={} qty={}", price, qty);
          if (price > fair_ask) {
            LOG(ERROR) << fmt::format(
                "potential self trade detected askBook{}/{}, fair_ask={} "
                "request qty={}, current qty{}",
                price,
                qty,
                fair_ask,
                fire_qty,
                total_qty);
            break;
          }
          total_qty += qty;
          fire_price = price;
          if (total_qty >= fire_qty) break;
        }

        if (total_qty > 0) {
          LOG(INFO) << fmt::format("Aggressive BUY order {}/{}", fire_price, fire_qty);
          order_executor_->ForceFireBuyOrder(upd.timestamp(), fire_price, fire_qty, "");
        } else {
          LOG(ERROR) << "our order is already top of book, not firing";
        }
      }
      fire_aggressive_order_ = false;
    }
  }
}

void BasisSubStrategy::onAccountInfo(const OrderUpdate& upd) {
  order_executor_->onAccountInfo(upd);
}

void BasisSubStrategy::onAccountOrder(const OrderUpdate& upd) {
  // LOG(INFO) << fmt::format(
  //     "BasisSubStrategy::onAccountOrder time {} side {} price {} qty {}",
  //     upd.product_order_info().event().event_time(),
  //     upd.product_order_info().event().trade_side(),
  //     upd.product_order_info().event().fill_price(),
  //     upd.product_order_info().event().fill_qty());
  order_executor_->onAccountOrder(upd);
  // LOG(INFO) << "receive product_order info!"
  // << upd.product_order_info().event().DebugString();
  // LOG(INFO) << "OrderContext"
  // << upd.product_order_info().order_contexts().size();
  // if (upd.product_order_info().event().type() == OrderEvent::ORDER_FILLED) {
  //   ManageTotalRisk(upd.timestamp());
  // }
}

void BasisSubStrategy::FireAggressiveOrderOnNextBook() {
  LOG(INFO) << fmt::format(
      "{} Will fire aggressive order on next book feed prev:{}",
      strategy_key_,
      fire_aggressive_order_);
  fire_aggressive_order_ = true;
}

void BasisSubStrategy::onOrderLog(const OrderGatewayLog& log) {
  if (auto* reporter = master_strategy_->GetStrategyReporter(reporter_config_); reporter ) {
    reporter->GetMutableStrategyLogger()->WriteOrderLog(log, true, GetCurrentTimestamp());
  }
}

double BasisSubStrategy::GetPosition() const {
  if (is_spot_){
    double extra_position = position_ - (max_position_ + min_position_) / 2;
    double spot_price = master_strategy_->GetLastMid(strategy_key_);
    return extra_position * spot_price / contract_value_;
  } else {
    return position_;
  }
}

void BasisSubStrategy::LogCurrentStatus() const {
  LOG(INFO) << fmt::format("executor min{} max{} ", previous_min_, previous_max_);
  order_executor_->LogStatus();
}

void BasisSubStrategy::ManageTotalRisk(int64_t timestamp) {
  double extra_long_postion =
      std::max(0.0, master_strategy_->GetMaxPosition() - master_strategy_->GetPosition());
  double new_max_position = std::min(max_position_, position_ + extra_long_postion);
  double extra_short_position =  // This is a position number
      std::max(0.0, master_strategy_->GetMaxPosition() + master_strategy_->GetPosition());
  double new_min_position = std::max(-max_position_, position_ - extra_short_position);
  if (previous_max_ != new_max_position) {
    LOG(INFO) << fmt::format(
        "{} Maximum updated: {} -> {}, extra_long_postion{} ",
        strategy_key_,
        previous_max_,
        new_max_position,
        extra_long_postion);
    previous_max_ = new_max_position;
  }
  if (previous_min_ != new_min_position) {
    previous_min_ = new_min_position;
    LOG(INFO) << fmt::format(
        "{} Minimum updated: {} -> {}, extra_short_position{}",
        strategy_key_,
        previous_min_,
        new_min_position,
        extra_short_position);
  }
  try {
    order_executor_->UpdateMinMaxPosition(new_min_position, new_max_position, timestamp);
  } catch (const std::out_of_range& e) {
    LOG(INFO) << "order executor position not ready" << e.what();
  };
}

bool BasisSubStrategy::SyncPosition() {
  bool position_updated = false;
  try {
    if (double latest_pos = order_executor_->GetPosition() + simulation_starting_position_;
        abs(position_ - latest_pos) > EPS) {
      LOG(INFO) << fmt::format(
          "{} Updated Position from order_executor_: {}->{}",
          strategy_key_,
          position_,
          latest_pos);
      last_trade_side_ =
          latest_pos > position_ ? TradeSide::TRADE_BUY_SIDE : TradeSide::TRADE_SELL_SIDE;
      position_ = latest_pos;
      position_updated = true;
    }
  } catch (const std::out_of_range& e) {
    LOG(INFO) << "order executor position not ready" << e.what();
  }
  return position_updated;
}

BasisStrategy::BasisStrategy(const path& config_json, int64_t timestamp)
    : driver_(nullptr), last_log_ts_(timestamp), position_(0) {
  LOG(INFO) << "BasisStrategy initializing";
  // TODO(Jingyuan): Make the parser work
  std::ifstream in(config_json);
  CHECK_THROW(in.is_open());
  nlohmann::json strategy_config = nlohmann::json::parse(in);

  // init conifgs
  lot_size_ = strategy_config["lot_size"].get<int>();
  edge_ = strategy_config["edge"].get<double>() * BPS;
  max_lean_bps_ = strategy_config["max_lean_bps"].get<double>() * BPS;

  //max position in USD exposure
  max_position_ = strategy_config["max_position"].get<double>();
  twma_.ResizeWindow(NS * strategy_config["window_period_seconds"].get<int>());

  // create two strategies
  for (auto [subreq, config] : strategy_config["exchange_config"].items()) {
    auto mea = MarketExchangeApi::FromString(subreq);
    auto product_str = config["product"].get<std::string>();
    std::unique_ptr<IProduct> product = CreateProductFromNormString(mea, product_str, timestamp);
    PassiveExecutorConfig exe_config;
    exe_config.set_mea(subreq);
    exe_config.set_symbol(product->symbol());
    exe_config.set_order_update_period(
        strategy_config["order_update_period_sec"].get<double>() * NS);
    exe_config.set_max_posting_period(strategy_config["max_posting_period_sec"].get<double>() * NS);
    if (strategy_config.contains("sticky_bp") && strategy_config["sticky_bp"].get<double>() > 0) {
      exe_config.set_sticky_bp(strategy_config["sticky_bp"].get<double>());
    }
    exe_config.set_lot_size(strategy_config["lot_size"].get<int>());
    exe_config.set_max_working_cnt(strategy_config["max_working_cnt"].get<int>());
    if (strategy_config.contains("leverage")) {
      exe_config.set_leverage(strategy_config["leverage"].get<double>());
    }

    if (strategy_config.contains("enable_aggressive_order") &&
        strategy_config["enable_aggressive_order"].get<bool>()) {
      aggressive_order_cooldown_ =
          strategy_config["aggressive_order_cooldown_sec"].get<double>() * NS;
      enable_aggressive_order_ = true;
    } else {
      enable_aggressive_order_ = false;
    }
    const std::string& strategy_key = GetProductKey(product.get());
    strategy_keys.push_back(strategy_key);
    strategies.insert({strategy_key,
                       std::make_unique<BasisSubStrategy>(
                           this,
                           config,
                           exe_config,
                           std::move(product),
                           timestamp)});

  }
  // have exactly 2 strategies
  CHECK_EQ_THROW(strategies.size(), 2);
}

void BasisStrategy::Init(Driver* driver) {
  driver_ = driver;
  auto driver_config = driver->config();
  for (auto& [name, config] : driver_config.feed().exchanges()) {
    (void)config;
    CHECK_THROW(std::count(strategy_keys.begin(), strategy_keys.end(), name))
        << "Feed config name must match the product keys: "
        << fmt::format("expected {} or {} got {}", strategy_keys[0], strategy_keys[1], name);
  }
  for (auto& [name, config] : driver_config.order().exchanges()) {
    (void)config;
    CHECK_THROW(std::count(strategy_keys.begin(), strategy_keys.end(), name))
        << "Order config name must match the product keys: "
        << fmt::format("expected {} or {} got {}", strategy_keys[0], strategy_keys[1], name);
  }

  for (auto& [strategy_key, strategy] : strategies) {
    (void)strategy_key;
    strategy->Init(driver);
  }

  for (auto& [strategy_key, strategy] : strategies) {
    (void)strategy_key;
    auto* reporter = GetStrategyReporter(strategy->reporter_config());
    reporter->RegisterProduct(strategy->product());
  }
}

void BasisStrategy::onBookFeed(const FeedUpdate& upd) {
  const auto & book = upd.book();
  const auto& key = upd.system_name();
  last_mids[key] = 0.5 * (book.Ask0()->price + book.Bid0()->price);
  if (last_mids.size() == 2) {
    twma_.Push(upd.timestamp(), last_mids[strategy_keys[0]] - last_mids[strategy_keys[1]]);
  }
  // master position will be updated on sub stgy book feed
  strategies[key]->onBookFeed(upd);

  if (upd.timestamp() - last_log_ts_ > LOGGING_INTERVAL) {
    last_log_ts_ = upd.timestamp();
    LogCurrentStatus();
  }

  if (enable_aggressive_order_ &&
      (upd.timestamp() - last_aggressive_order_timestamp_ > aggressive_order_cooldown_)) {
    if (last_aggressive_order_timestamp_ == 0) {
      LOG(INFO) << "first book feed, not fire aggressive order";
    } else if (abs(position_) < EPS) {
      LOG(INFO) << "total position is 0 No need to fire aggressive order";
    } else {
      auto target_side = position_ > 0 ? TradeSide::TRADE_SELL_SIDE : TradeSide::TRADE_BUY_SIDE;
      LOG(INFO) << fmt::format(
          "will fire aggressive order to clear current position {}",
          position_);
      for (auto& [key, strategy] : strategies) {
        LOG(INFO) << fmt::format(
            "{} last side {}",
            key,
            strategy->last_trade_side_);
        if (strategy->last_trade_side_ == target_side) {
          strategy->FireAggressiveOrderOnNextBook();
          break;
        }
      }
    }
    last_aggressive_order_timestamp_ = upd.timestamp();
  }
  // printf("%ld,%ld,book,%s,%s,%f,%f,%f,%f,0,0\n", upd.timestamp(),
  //        book.Timestamp(), upd.native_symbol().c_str(),
  //        ExchangeType_Name(upd.exchange()).c_str(), book.Ask0()->price,
  //        book.Ask0()->qty, book.Bid0()->price, book.Bid0()->qty);
}

void BasisStrategy::onTradeFeed(const FeedUpdate& upd) {
  // auto& trade = upd.trade();
  // printf("%ld,%ld,trade,%s,%s,0,0,0,0,%f,%f\n", upd.timestamp(),
  //        trade.timestamp, upd.native_symbol().c_str(),
  //        ExchangeType_Name(upd.exchange()).c_str(), trade.price,
  //        trade.fill_qty);
}

void BasisStrategy::onAccountOrder(const OrderUpdate& upd) {
  if (upd.product_order_info().event().type() == OrderEvent::ORDER_FILLED) {
    BasisSubStrategy* strategy = strategies[upd.system_name()].get();
    strategy->onAccountOrder(upd);
    SyncPositionAndRisk(upd.timestamp());
  }
}

void BasisStrategy::onAccountInfo(const OrderUpdate& upd) {
  SyncPositionAndRisk(upd.timestamp());
  BasisSubStrategy* strategy = strategies[upd.system_name()].get();
  strategy->onAccountInfo(upd);
  GetStrategyReporter(strategy->reporter_config())->UpdateMidp(strategy->product(), last_mids[upd.system_name()]);
  GetStrategyReporter(strategy->reporter_config())->onAccountInfo(upd);
}

void BasisStrategy::onOrderLog(const OrderGatewayLog& log) {
  // TODO(jingyuan) : fix me in case of futures/futures has same keys
  if (strategy_keys[0].find(log.account_request().exchange())!=std::string::npos
    && strategy_keys[0].find(log.account_request().market_type())!=std::string::npos) {
     strategies[strategy_keys[0]]->onOrderLog(log);
  }else{
    strategies[strategy_keys[1]]->onOrderLog(log);
  }


}

void BasisStrategy::onCleanup(Driver*  /*driver*/) {
  LOG(INFO) << "BasisStrategy cancelling all working orders";
  for (auto& [key, strategy] : strategies) {
    (void)key;
    strategy->GetMutableExecutor()->ForceCancelAllWorkingOrders();
  }
}

std::optional<std::pair<double, double>> BasisStrategy::fair_value(const std::string& key) {
  if (twma_.last_popped()) {
    double ref_mid = last_mids[otherStrategy(key)];
    double lean = -(position_ / max_position_) * max_lean_bps_;
    double direction_multiplier = key == strategy_keys[0] ? 1 : -1;
    ref_mid += twma_.tw_mean() * direction_multiplier;
    ref_mid *= (1 + lean);
    return std::make_pair(ref_mid * (1 - edge_), ref_mid * (1 + edge_));
  } else {
    return {};
  }
}

void BasisStrategy::LogCurrentStatus() {
  LOG(INFO) << "Total Position: " << position_;
  LOG(INFO) << fmt::format("twma: mean {}; size {}", twma_.tw_mean(), twma_.size());
  for (const std::string& key : strategy_keys) {
    if (const auto& results = fair_value(key); results) {
      auto [bid, ask] = results.value();
      strategies[key]->LogCurrentStatus();
      LOG(INFO) << fmt::format(
          "{} position={} mid={} fair_bid {}, ask{} minpos {} maxpos{}",
          key,
          strategies[key]->GetPosition(),
          last_mids[key],
          bid,
          ask,
          strategies[key]->GetMinPosition(),
          strategies[key]->GetMaxPosition());
      LOG(INFO) << strategies[key]->order_system()->gateway()->order_manager().DebugString();
    }
  }
}

StrategyReporter* BasisStrategy::GetStrategyReporter(const StrategyManagerConfig& config) {
  const std::string& strategy_name = config.strategy_name();
  if (strat_reporters_.find(strategy_name) == strat_reporters_.end()) {
    auto reserve = reserves_.find(strategy_name);
    CHECK(reserve != reserves_.end())
        << "Reserve is not initialized for " << strategy_name;
    strat_reporters_[strategy_name] = std::make_unique<StrategyReporter>(config, reserve->second);
  }
  return strat_reporters_[strategy_name].get();
}

std::unordered_map<std::string, double>& BasisStrategy::GetReserve(const StrategyManagerConfig& config) {
  const std::string& strategy_name = config.strategy_name();
  if (reserves_.find(strategy_name) == reserves_.end()) {
    reserves_.emplace(strategy_name, std::unordered_map<std::string, double>{});
  }
  return reserves_[strategy_name];
}

void BasisStrategy::SyncPositionAndRisk(int64_t timestamp) {
  bool updated = false;
  for (const std::string& key : strategy_keys) {
    updated |= strategies[key]->SyncPosition();
  }
  if (updated) {
    position_ = 0;
    for (const std::string& key : strategy_keys) {
      position_ += strategies[key]->GetPosition();
    }
    LOG(INFO) << "Sync new master position: " << position_;
    for (const std::string& key : strategy_keys) {
      strategies[key]->ManageTotalRisk(timestamp);
    }
  }
}

}  // namespace coin2::strategy::basis_strategy
